package com.yanggu.flink.datastream_api.multi_stream_transform.combine_stream

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.CoGroupFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import java.lang

/**
 * CoGroup Join是Window Join的底层依赖
 * Window Join只能实现INNER JOIN
 * CoGroup Join可以实现任意LEFT JOIN、RIGHT JOIN、FULL JOIN等
 *
 * 输出
 * [(a,1000), (a,2000)]=>[(a,3000), (a,4000)]
 * [(b,1000), (b,2000)]=>[(b,3000), (b,4000)]
 *
 */
object CoGroupDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val dataStream1 = env
      .fromElements(
        ("a", 1000L),
        ("b", 1000L),
        ("a", 2000L),
        ("b", 2000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(element: (String, Long), recordTimestamp: Long) = element._2
        })
      )

    val dataStream2 = env
      .fromElements(
        ("a", 3000L),
        ("b", 3000L),
        ("a", 4000L),
        ("b", 4000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(element: (String, Long), recordTimestamp: Long) = element._2
        })
      )

    dataStream1
      .coGroup(dataStream2)
      .where(tuple1 => tuple1._1)
      .equalTo(tuple2 => tuple2._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5L)))
      .apply(new CoGroupFunction[(String, Long), (String, Long), String] {

        //first: 第一条流窗口关闭后的数据list
        //second: 第二条流窗口关闭后的数据list
        override def coGroup(first: lang.Iterable[(String, Long)], second: lang.Iterable[(String, Long)], out: Collector[String]): Unit = {
          out.collect(first + "=>" + second)
        }
      })
      .print()

    env.execute()

  }

}
